package com.gvtv.main.execute;

import java.util.Date;
import java.util.List;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.yarn.api.records.ApplicationReport;
import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
import org.apache.log4j.Logger;
import org.joda.time.DateTime;

import com.gvtv.main.exception.ExecuteException;
import com.gvtv.main.execute.ExecuteJob.ExecuteJobState;
import com.gvtv.main.hadoop.MapReduceRun;
import com.gvtv.main.hadoop.yarn.YarnHelper;
import com.gvtv.main.schedul.SchedulEnv;
import com.gvtv.main.schedul.SchedulInfo;
import com.gvtv.main.util.Constant;

public class MapReduceJobWorker extends AbstractJobWorker implements JobWorker{

	private Logger logger = Logger.getLogger(MapReduceJobWorker.class);
	
	public MapReduceJobWorker(SchedulInfo schedulInfo, ExecuteJob executeJob, SchedulEnv env) {
		super(schedulInfo, executeJob, env);
	}

	@Override
	public void work() {
		job.setStartTime(new Date());
		logger.info("  job ["+job.getId()+"]开始运行，运行时间："+new DateTime(job.getSubmitTime().getTime()).toString(Constant.DATETIME_FORMAT));
		env.convert(job);
		MapReduceRun mrRun = new MapReduceRun();
		YarnHelper yarnHelper = new YarnHelper();
		try {
			if(StringUtils.isBlank(job.getMapReduceJobName())){
				throw new ExecuteException("job ["+job.getId()+"]的MAPREDUCE程序的作业名称[mapreduceName]参数为空 。");
			}
			mrRun.run(job.getMRRunArgs().toArray(new String[]{}));
			List<ApplicationReport> list = yarnHelper.getAppByName(job.getMapReduceJobName());
			if(list==null || list.isEmpty()){
				job.setJobState(ExecuteJobState.FAILED);
				logger.error("根据job["+job.getId()+"] 的["+job.getMapReduceJobName()+"]没有找到对应运行应用，无法判断job的最后运行状态。");
			}else {
				if(list.size()>1){
					job.setJobState(ExecuteJobState.FAILED);
					logger.error("根据job["+job.getId()+"] 的["+job.getMapReduceJobName()+"]找到了多个对应运行应用，无法判断job的最后运行状态。");
				}else {
					if(list.get(0).getFinalApplicationStatus()==FinalApplicationStatus.SUCCEEDED){
						job.setJobState(ExecuteJobState.SUCCESS);
					}else {
						job.setJobState(ExecuteJobState.FAILED);
					}
				}
			}
		} catch (Throwable e) {
			logger.error("job ["+job.getId()+"]运行失败。" + e);
			job.setJobState(ExecuteJobState.FAILED);
		} finally {
			yarnHelper.close();
		}
		job.setEndTime(new Date());
		logger.info("job ["+job.getId()+"]结束运行，结束时间："+new DateTime(job.getEndTime().getTime()).toString(Constant.DATETIME_FORMAT)+"，运行状态，"+job.getJobState());
	}
}
